package com.gzs.learn.payment.failover;

import com.gzs.learn.common.util.JsonUtil;
import com.gzs.learn.common.util.SpringContextUtil;
import com.gzs.learn.payment.config.ThreadPoolConfig;
import com.gzs.learn.payment.config.properties.FailoverProperties;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import org.springframework.data.redis.core.HashOperations;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.ZSetOperations;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.stereotype.Component;

import java.util.Map;
import java.util.Map.Entry;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.ReentrantLock;

/**
 * @author zhisong.guan
 */
@Slf4j
@Component
@SuppressWarnings({"rawtypes", "unchecked"})
@RequiredArgsConstructor
public class FailoverService {
    /**
     * 间隔多久后重试,以分钟为单位
     */
    private static final int[] DEFAULT_FAILOVER_SEQ = {1, 5, 10, 15, 30, 60, 60 * 3, 60 * 12};
    private static final String ZSET_NAME = "-zset-";
    private static final String HASH_NAME = "-hash-meta";
    private static final AtomicInteger KEY = new AtomicInteger(0);
    private volatile boolean isInit = false;
    private final ThreadPoolTaskExecutor pool = SpringContextUtil.getBean(ThreadPoolConfig.POOL_NAME_ASYNC);
    private TimeUnit timeUnit = null;
    private final ReentrantLock mainLock = new ReentrantLock();
    private int[] seq = null;

    private final ConcurrentHashMap<Integer, FailoverHandler> registers = new ConcurrentHashMap<>();

    private final FailoverProperties failoverProperties;

    private final RedisTemplate<Object, Object> redisTemplate;

    public Object doFailOver(FailoverHandler handler, Object param) {
        if (StringUtils.isBlank(failoverProperties.getSetPrefix())) {
            throw new IllegalArgumentException("failover set prefix must config");
        }
        // 初始化seq
        this.initConfig();
        // 注册处理器
        int bizKey = this.registerHandler(handler);

        try {
            // 无异常直接返回
            return handler.handle(param);
        } catch (Throwable e) {
            try {
                // 第一次失败立即发起一次重试
                return handler.handle(param);
            } catch (Throwable e1) {
                String firstSetName = this.createZsetName(bizKey, 0);
                long firstSetScore = System.currentTimeMillis() + timeUnit.toMillis(seq[0]);
                // 丢入重试队列
                ZSetOperations<Object, Object> opsForzSet = redisTemplate.opsForZSet();
                FailoverData data = new FailoverData(bizKey, param);
                String zsetData = JsonUtil.toJSONString(data);
                opsForzSet.add(firstSetName, zsetData, firstSetScore);
                return null;
            }
        }
    }

    public void doRetry() {
        this.initConfig();
        int currentBizKeyNum = KEY.get();
        // 业务类型数量
        for (int i = 0; i < currentBizKeyNum; i++) {
            int bizKey = i;
            pool.submit(() -> this.doSingleBiz(bizKey, seq));
        }
    }

    /**
     * 处理单个业务类型的数据
     */
    private void doSingleBiz(int bizKey, int[] seq) {
        long score = System.currentTimeMillis();
        // 重试队列
        for (int i = 0; i < seq.length; i++) {
            this.handleSingleZset(bizKey, i, score, false);
        }
        // 处理最后一条队列内容
        int last = seq.length;
        this.handleSingleZset(bizKey, last, score, true);
    }

    /**
     * 处理单个zset队列数据
     *
     * @param bizKey
     * @param index
     * @param score
     * @param isLast
     */
    private void handleSingleZset(int bizKey, int index, long score, boolean isLast) {
        FailoverHandler handler = registers.get(bizKey);
        int nextIndex = index + 1;
        String current = this.createZsetName(bizKey, index);
        String next = this.createZsetName(bizKey, nextIndex);
        ZSetOperations<Object, Object> opsForzSet = redisTemplate.opsForZSet();
        Set<Object> dataSet = opsForzSet.rangeByScore(current, 0, score);
        if (CollectionUtils.isEmpty(dataSet)) {
            return;
        }
        for (Object data : dataSet) {
            FailoverData failoverData = JsonUtil.parseObject(String.valueOf(data), FailoverData.class);
            Object obj = JsonUtil.parseObject(failoverData.getJson(), failoverData.getCls());
            try {
                handler.handle(obj);
                if (log.isDebugEnabled()) {
                    log.info("zset:{},data:{},retry success until times:{}", current, data, index);
                }
            } catch (Throwable e) {
                if (isLast) {
                    handler.onFail(obj);
                } else {
                    long nextScore = score + timeUnit.toMillis(seq[index]);
                    opsForzSet.add(next, data, nextScore);
                }
            }
        }
    }

    private void initConfig() {
        if (isInit) {
            return;
        }
        ReentrantLock lock = mainLock;
        lock.lock();
        try {
            String seqExp = failoverProperties.getSeq();
            if (StringUtils.isNotBlank(failoverProperties.getSeq())) {
                String[] strs = seqExp.split(",");
                String type = strs[0];
                // 类型
                if ("ms".equalsIgnoreCase(type)) {
                    timeUnit = TimeUnit.MILLISECONDS;
                } else if ("s".equalsIgnoreCase(type)) {
                    timeUnit = TimeUnit.SECONDS;
                } else if ("m".equals(type)) {
                    timeUnit = TimeUnit.MINUTES;
                } else if ("h".equalsIgnoreCase(type)) {
                    timeUnit = TimeUnit.HOURS;
                }
                seq = new int[strs.length - 1];
                for (int i = 0; i < seq.length; i++) {
                    // due to strs[0] is type
                    seq[i] = Integer.parseInt(strs[i + 1].trim());
                }
            } else {
                // default settings
                timeUnit = TimeUnit.MINUTES;
                seq = DEFAULT_FAILOVER_SEQ;
            }
            isInit = true;
        } finally {
            lock.unlock();
        }
    }

    private int registerHandler(FailoverHandler handler) {
        int bizKey = -1;
        if (!registers.containsValue(handler)) {
            bizKey = KEY.getAndIncrement();
            registers.put(bizKey, handler);
            // 向meta中写入信息
            String hashName = failoverProperties.getSetPrefix() + HASH_NAME;
            HashOperations<Object, Object, Object> opsForHash = redisTemplate.opsForHash();
            for (Entry<Integer, FailoverHandler> entry : registers.entrySet()) {
                String className = entry.getValue().getClass().getName();
                // 简单处理掉代理类
                if (className.indexOf('$') > 0) {
                    className = className.substring(0, className.indexOf('$'));
                }
                opsForHash.put(hashName, entry.getKey() + "", className);
            }
        } else {
            for (Entry<Integer, FailoverHandler> entry : registers.entrySet()) {
                if (entry.getValue() == handler) {
                    bizKey = entry.getKey();
                    break;
                }
            }
        }
        return bizKey;
    }

    private String createZsetName(int bizKey, int setIndex) {
        return failoverProperties.getSetPrefix() + "-" + bizKey + ZSET_NAME + setIndex;
    }

    /**
     * 初始化handler
     */
    public void initHandlers() {
        ReentrantLock lock = mainLock;
        lock.lock();
        try {
            String hashName = failoverProperties.getSetPrefix() + HASH_NAME;
            HashOperations<Object, Object, Object> opsForHash = redisTemplate.opsForHash();
            Map<Object, Object> clsMap = opsForHash.entries(hashName);

            for (Entry<Object, Object> entry : clsMap.entrySet()) {
                String className = String.valueOf(entry.getValue());
                if (className.indexOf('$') > 0) {
                    className = className.substring(0, className.indexOf('$'));
                }
                try {
                    Class<?> cls = Class.forName(className);
                    FailoverHandler handler = (FailoverHandler) SpringContextUtil.getBean(cls);
                    if (registers.containsValue(handler)) {
                        continue;
                    }

                    int bizKey = KEY.getAndIncrement();
                    registers.put(bizKey, handler);
                } catch (ClassNotFoundException e) {
                    log.warn("can not find handler with class:{}", className);
                } catch (Exception e) {
                    log.error("get handler {} failed due exception", className, e);
                }
            }
        } finally {
            lock.unlock();
        }
    }
}
